package com.ziroom.crm.mq.rabbit.spring.core;

import com.alibaba.fastjson.JSON;
import com.ziroom.crm.mq.core.MqType;
import com.ziroom.crm.mq.core.config.MqConfig;
import com.ziroom.crm.mq.core.exception.MqConsumeException;
import com.ziroom.crm.mq.core.logging.Log;
import com.ziroom.crm.mq.core.logging.LogFactory;
import com.ziroom.crm.mq.core.reconsume.ReConsume;
import com.ziroom.crm.mq.core.reconsume.ReConsumeMethodHolder;
import com.ziroom.crm.mq.core.support.ProcessorInvokerDelegate;
import com.ziroom.crm.mq.rabbit.spring.client.ReConsumeChannel;
import com.ziroom.crm.mq.rabbit.spring.data.RabbitMessage;
import com.ziroom.crm.mq.rabbit.spring.resolve.RabbitResolve;
import com.ziroom.crm.mq.rabbit.spring.wrapper.RabbitConsumeQueueWrapper;
import com.ziroom.crm.rock.spring.util.SpringContextUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;

import java.nio.charset.Charset;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 重新消费
 *
 * @author WeiShaoying
 * @version 1.0
 * @date Created in 2020年09月07日 11:01
 * @since 1.0
 */
@Slf4j
public class RabbitMqReConsume implements ReConsume<RabbitMessage> {

    //volatile
    private static final AtomicBoolean INITIALIZED = new AtomicBoolean(false);

    private static final ReConsumeChannel RE_CONSUME_CHANNEL = new ReConsumeChannel();

    private ProcessorInvokerDelegate facade;

    @Override
    public MqType getType() {
        return MqType.RABBIT;
    }

    @Override
    public void reconsume(RabbitMessage rabbitMessage) {
        init();
        RabbitConsumeQueueWrapper rabbitQueueWrapper = RabbitResolve.buildConsumeQueueWrapper(rabbitMessage);
        //执行前置处理
        facade.preReConsumeInvoke(rabbitQueueWrapper, rabbitMessage);

        try {
            MessageProperties messageProperties = JSON.parseObject(rabbitMessage.getMessageProperties(), MessageProperties.class);
            byte[] body = rabbitMessage.getMessageBody().getBytes(Charset.forName(MqConfig.charset()));
            // 构建Spring的消息
            Message message = new Message(body, messageProperties);
            // 设置retryCount，让下游知道来源与持久化重试逻辑，以及当前重试次数
            RabbitResolve.setReconsumeCount(message, rabbitMessage.getRetryCount());

            Object object = ReConsumeMethodHolder.getObject(rabbitQueueWrapper);

            if (object instanceof ChannelAwareMessageListener) {
                log.debug("ChannelAwareMessageListener onMessage()");
                ((ChannelAwareMessageListener) object).onMessage(message, RE_CONSUME_CHANNEL);
            } else if (object instanceof MessageListener) {
                // 执行
                log.debug("MessageListener onMessage()");
                ((MessageListener) object).onMessage(message);
            } else {
                log.error("[crm-mq]未找到消息处理类：queueName:" + rabbitMessage.getQueueName());
                //重新执行失败处理
                throw new MqConsumeException("未找到处理类：queueName:" + rabbitMessage.getQueueName());
            }
        } catch (Exception e) {
            log.error("[crm-mq]重新消费失败,messageId:" + rabbitMessage.getId() + ".", e);
            rabbitMessage.setException(e);
        } finally {
            // 执行重发后置处理(计算再执行时间+持久化)
            facade.postReConsumeInvoke(rabbitQueueWrapper, rabbitMessage);
        }
    }

    private void init() {
        if (!INITIALIZED.compareAndSet(false, true)) {
            return;
        }
        facade = SpringContextUtils.getBean(ProcessorInvokerDelegate.class);
    }

}
